package org.timewindow

import org.FlinkStreamApp
import org.apache.flink.api.common.state.ValueStateDescriptor
import org.apache.flink.api.scala.typeutils.Types
import org.apache.flink.streaming.api.functions.co.CoProcessFunction
import org.apache.flink.streaming.api.scala._
import org.apache.flink.util.Collector
import org.bean.SensorReading
import org.diysource.SensorSource

/**
 * description ：join 流- id 为 2 的传感器读数放行 10s
 * author      ：剧情再美终是戏 
 * mail        : 13286520398@163.com
 * date        ：Created in 2020/2/23 11:23
 * modified By ：
 * version:    : 1.0
 */
object CoProcessFunctionExample extends FlinkStreamApp {
  override def doSomeThing(environment: StreamExecutionEnvironment) = {
    // 传感器读数
    val readings = environment.addSource(new SensorSource)

    // 开关流
    val filterSwitches = environment.fromCollection(Seq(
      ("2", 2 * 1000L) // id 为 2 的传感器读数放行 10s
    ))

    val forwardedReadings = readings.keyBy(_.id).connect(filterSwitches.keyBy(_._1)).process(new ReadingFilter)

    forwardedReadings.print()
  }

  // IN1, IN2, OUT
  class ReadingFilter extends CoProcessFunction[SensorReading, (String, Long), SensorReading] {
    // 设置状态标记位
    lazy val openSwith = getRuntimeContext.getState(
      new ValueStateDescriptor[Boolean]("openSwith", Types.of[Boolean])
    )
    // 保存定时器时间
    lazy val timer = getRuntimeContext.getState(
      new ValueStateDescriptor[Long]("timer", Types.of[Long])
    )

    override def processElement1(value: SensorReading, ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#Context, out: Collector[SensorReading]) = {
      if (openSwith.value()) {
        // 往下写数据
        out.collect(value)
      }
    }

    override def processElement2(value: (String, Long), ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#Context, out: Collector[SensorReading]) = {
      // 当开关流的数据过来了，
      // 1、设置开关为打开状态
      openSwith.update(true)

      // 2、设置定时器，时间到了就由定时器去关闭 获取形状流的时间，设置定时器的时间
      val times = ctx.timerService().currentProcessingTime() + value._2
      ctx.timerService().registerProcessingTimeTimer(times)
      timer.update(times)
    }

    override def onTimer(timestamp: Long, ctx: CoProcessFunction[SensorReading, (String, Long), SensorReading]#OnTimerContext, out: Collector[SensorReading]) = {
      // 定时器到了时间，就清理开关还有清理定时器
      openSwith.clear()
      timer.clear()
    }
  }

}
